package drds.binlog.metadata;

import com.google.common.base.Function;
import com.google.common.collect.ComputingMap;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.Position;
import drds.binlog.common.position.PositionRange;
import org.springframework.util.Assert;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 组合memory + zookeeper的使用模式
 */

public class MixedMetaDataManager extends MemoryMetaDataManager implements MetaDataManager
{

    @SuppressWarnings("serial")
    private final Position nullCursor = new Position()
    {
    };
    private ExecutorService executor;
    private ZooKeeperMetaDataManager zooKeeperMetaManager;

    public void start()
    {
        super.start();
        Assert.notNull(zooKeeperMetaManager);
        if (!zooKeeperMetaManager.isRunning())
        {
            zooKeeperMetaManager.start();
        }

        executor = Executors.newFixedThreadPool(1);
        destinationToClientInfoListMap = ComputingMap.makeComputingMap(new Function<String, List<ClientInfo>>()
        {

            public List<ClientInfo> apply(String destination)
            {
                return zooKeeperMetaManager.getClientInfoList(destination);
            }
        });

        clientInfoToPositionMap = ComputingMap.makeComputingMap(new Function<ClientInfo, Position>()
        {

            public Position apply(ClientInfo clientInfo)
            {
                Position position = zooKeeperMetaManager.getPosition(clientInfo);
                if (position == null)
                {
                    return nullCursor; // 返回一个空对象标识，避免出现异常
                } else
                {
                    return position;
                }
            }
        });

        clientInfoToBatchIdToPositionRangeMapMap = ComputingMap.makeComputingMap(new Function<ClientInfo, BatchIdToPositionRangeConcurrentMap>()
        {

            public BatchIdToPositionRangeConcurrentMap apply(ClientInfo clientInfo)
            {
                // 读取一下zookeeper信息，初始化一次
                BatchIdToPositionRangeConcurrentMap batches = BatchIdToPositionRangeConcurrentMap.create(clientInfo);
                Map<Long, PositionRange> positionRanges = zooKeeperMetaManager.getBatchIdToPositionRangeMap(clientInfo);
                for (Map.Entry<Long, PositionRange> entry : positionRanges.entrySet())
                {
                    batches.addPositionRange(entry.getKey(), entry.getValue()); // 添加记录到指定batchId
                }
                return batches;
            }
        });
    }

    public void stop()
    {
        super.stop();

        if (zooKeeperMetaManager.isRunning())
        {
            zooKeeperMetaManager.stop();
        }

        executor.shutdownNow();
        destinationToClientInfoListMap.clear();
        clientInfoToBatchIdToPositionRangeMapMap.clear();
    }

    public void subscribe(final ClientInfo clientInfo) throws MetaDataManagerException
    {
        super.subscribe(clientInfo);

        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.subscribe(clientInfo);
            }
        });
    }

    public void unsubscribe(final ClientInfo clientInfo) throws MetaDataManagerException
    {
        super.unsubscribe(clientInfo);

        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.unsubscribe(clientInfo);
            }
        });
    }

    public void updatePosition(final ClientInfo clientInfo, final Position position)
            throws MetaDataManagerException
    {
        super.updatePosition(clientInfo, position);

        // 异步刷新
        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.updatePosition(clientInfo, position);
            }
        });
    }

    @Override
    public Position getPosition(ClientInfo clientInfo) throws MetaDataManagerException
    {
        Position position = super.getPosition(clientInfo);
        if (position == nullCursor)
        {
            return null;
        } else
        {
            return position;
        }
    }

    public Long addPositionRange(final ClientInfo clientInfo, final PositionRange positionRange)
            throws MetaDataManagerException
    {
        final Long batchId = super.addPositionRange(clientInfo, positionRange);
        // 异步刷新
        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.addPositionRange(clientInfo, batchId, positionRange);
            }
        });
        return batchId;
    }

    public void addPositionRange(final ClientInfo clientInfo, final Long batchId, final PositionRange positionRange)
            throws MetaDataManagerException
    {
        super.addPositionRange(clientInfo, batchId, positionRange);
        // 异步刷新
        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.addPositionRange(clientInfo, batchId, positionRange);
            }
        });
    }

    public PositionRange removePositionRange(final ClientInfo clientInfo, final Long batchId)
            throws MetaDataManagerException
    {
        PositionRange positionRange = super.removePositionRange(clientInfo, batchId);
        // 异步刷新
        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.removePositionRange(clientInfo, batchId);
            }
        });

        return positionRange;
    }

    public void clearBatchIdToPositionRangeMap(final ClientInfo clientInfo) throws MetaDataManagerException
    {
        super.clearBatchIdToPositionRangeMap(clientInfo);

        // 异步刷新
        executor.submit(new Runnable()
        {

            public void run()
            {
                zooKeeperMetaManager.clearBatchIdToPositionRangeMap(clientInfo);
            }
        });
    }

    // =============== setter / getter ================
    public void setZooKeeperMetaManager(ZooKeeperMetaDataManager zooKeeperMetaManager)
    {
        this.zooKeeperMetaManager = zooKeeperMetaManager;
    }
}
